package com.exd.flink.stream.flow;

import java.util.ArrayList;
import java.util.List;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 描述: 解析车辆数据
 * 创建人: 赵兴炎
 * 日期: 2022年9月13日
 */
public class VinStream {

	/**
	 * 解析车辆数据
	 */
	public static void analyticalData(DataStream<String> dataStreamValue) {

		SingleOutputStreamOperator<String> vinDataStream = dataStreamValue.map(new RichMapFunction<String, String>() {
			private static final long serialVersionUID = 1L;
			@Override
			public String map(String message) throws Exception {
				String[] split = message.split("\"vin\":\"");
				return split[1].split("\",\"")[0];
			}
		});
		
		vinDataStream.keyBy(data -> true).process(new KeyedProcessFunction<Boolean,String,String>(){
			private static final long serialVersionUID = 1L;
			
			MapState<String, String> vindataMap;
			
			@Override
			public void open(Configuration parameters) throws Exception {
				vindataMap = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("vin-data-map", String.class, String.class));
			}
			
			@Override
			public void processElement(String message, Context context,Collector<String> collector) throws Exception {
				vindataMap.put(message, message);
				context.timerService().registerProcessingTimeTimer(100000);
			}
			
			@Override
			public void onTimer(long timestamp, OnTimerContext ctx,Collector<String> out) throws Exception {
				List<String> vinList = new ArrayList<>();
				vindataMap.values().forEach(data ->{vinList.add(data);});
				vindataMap.clear();
				System.out.println(vinList.size());
			}
			
		});
	}

}
